package com.flute.haflute.jobbox.child;

import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.jobbox.base.RunProcessInterface;
import com.flute.haflute.jobbox.communication.messages.MessageBase;

/**
 * 任务调度后台线程 获得任务请求消息，进行任务环境初始化、任务调度和监控 子节点核心线程
 * 
 */
public class TaskTracker extends Thread implements RunProcessInterface {
	private static Logger logger = LoggerFactory.getLogger(TaskTracker.class);
	private boolean initiallized = false;
	private static TaskTracker instance = null;

	/** 存储任务请求 */
	private List<MessageBase> requests;
	TaskSchedulerManager schedulerManager;
	/** 系统是否关闭标记，如果关闭则准备推出 */
	private boolean shutdown = false;
	/** 系统推出时，是否先执行完所有任务，称为优雅的关闭 */
	private boolean gracefullyShutdonw = false;
	/** 任务请求列表requests的专用锁 */
	final private Object lockFlag = new Object();

	/**
	 * 唯一构造函数
	 */
	public TaskTracker() {
		super("TASKTRACKER");
		init();
	}

	/** 初始化环境参数 */
	private void init() {
		initiallized = false;
		logger.info("TaskTracker initializing...");

		schedulerManager = new TaskSchedulerManager();
		requests = new ArrayList<MessageBase>(10);

		instance = this;
		logger.info("TaskTracker started!!");
	}

	public static synchronized TaskTracker getInstance() {
		int totalTime = 1000;
		int trys = 0;
		// 如果JobTasker还没有完成初始化，等待一次，直到等待超时
		while ((instance == null || !instance.initiallized)
				&& (trys++ < totalTime)) {
			doWait();
		}

		// 等待超时后还没有完成初始化，什么环节出现了问题，
		// 先启动一个实例保障当前任务的正常进行
		if (instance == null) {
			instance = new TaskTracker();
			instance.start();
		}
		return instance;
	}

	/**
	 * 处理函数 从任务请求队列中取出任务请求，对请求进行分析 然后调度执行
	 */
	@Override
	public void run() {
		while (!shutdown || (gracefullyShutdonw && requests.size() > 0)) {
			if (requests.size() > 0) {
				logger.debug("execute once size={}", requests.size());

				List<MessageBase> requests2deal = null;
				synchronized (lockFlag) {
					requests2deal = requests;
					requests = new ArrayList<MessageBase>();
				}

				for (MessageBase oneReq : requests2deal) {
					schedulerManager.prepare2runTask(oneReq);
				} // for
			}
			doWait();
		}
	}

	/** 接受任务请求 */
	public void doNotify(MessageBase message) throws RemoteException {
		logger.info("Child: {} received 1 message:{}", ChildContext.getMyID(),
				message);
		synchronized (lockFlag) {
			requests.add(message);
		}
	}

	/** 接受任务请求 */
	public void doNotify(List<MessageBase> messages) throws RemoteException {
		logger.debug("Child: {} received message size={}",
				ChildContext.getMyID(), messages.size());
		synchronized (lockFlag) {
			requests.addAll(messages);
		}
	}

	@Override
	public void callMain(String... params) {
		if (instance != null) {
			instance.start();
			initiallized = true;
		} else
			logger.error("JobTracker not initialized, some unexpected error happened, please check");
	}

	/** 调度器等待一定时间，让出CPU */
	public static void doWait() {
		try {
			Thread.sleep(200L);
		} catch (Exception e) {
		}
	}

	@Override
	public boolean isInitialized() {
		return initiallized;
	}

	@Override
	public void shutdown() {
		shutdown = true;
		gracefullyShutdonw = true;
	}

	@Override
	public void restart() {
	}

}
